package com.sunday.common.mq.kafka.brave.autoconfiguration;

import brave.kafka.clients.KafkaTracing;
import brave.kafka.clients.TracingKafkaPostProcessor;
import com.sunday.common.mq.kafka.brave.aspectj.TracingKafkaAspect;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.KafkaTemplate;

/**
 * 补全 kafka pull 模式链路问题
 *
 * @author zsy
 * @since 2023/8/16
 */
@Slf4j
@AutoConfiguration(after = KafkaTemplate.class)
@ConditionalOnBean(KafkaTracing.class)
public class KafkaTracingAutoConfiguration {


    /**
     * 对应 consumer 拉取后立即关闭的情况进行 业务延续 接收后跟踪处理器
     *
     * @param kafkaTracing
     * @return {@link TracingKafkaPostProcessor}
     */
    @Bean
    @ConditionalOnMissingBean
    public TracingKafkaPostProcessor tracingAfterReceivePostProcessor(KafkaTracing kafkaTracing) {
        return new TracingKafkaPostProcessor(kafkaTracing, true);
    }

    /**
     * 进行 @TracingKafkaHandler/@KafkaListener/@KafkaHandler 标签的aop拦截，进行业务关闭处理
     *
     * @param tracingAfterReceivePostProcessor
     * @return {@link TracingKafkaAspect}
     */
    @Bean
    @ConditionalOnMissingBean
    public TracingKafkaAspect tracingReceiveAspect(TracingKafkaPostProcessor tracingAfterReceivePostProcessor) {
        return new TracingKafkaAspect(tracingAfterReceivePostProcessor);
    }

}
